package com.chentf.rabbitmq.workquene03;

import com.chentf.rabbitmq.utils.RabbitMqUtilsSGG;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

/**
 *发布确认模式，
 * 3、异步批量确认
 */
public class SyncTask3 {
    
    // 批量发消息的个数
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        //3、异步批量确认
        // 发布1000个异步确认消息，耗时36ms
        SyncTask3.publicMessageAsync();

    }

    private static void publicMessageAsync() throws IOException, TimeoutException {
        Channel channel = RabbitMqUtilsSGG.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,false,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long start = System.currentTimeMillis();
        //消费确认成功回调函数
        ConfirmCallback ackCallback = (deliveryTag,multiply) -> {
            System.out.println("确认的消息："+deliveryTag);
        };
        // 消息确认失败回调函数
        /*
         * 参数1：消息的标记
         * 参数2：是否为批量确认
         * */
        ConfirmCallback nackCallback = (deliveryTag,multiply) -> {
            System.out.println("未确认的消息："+deliveryTag);
        };
        // 准备消息的监听器，监听哪些消息成功，哪些消息失败
        /*
         * 参数1：监听哪些消息成功
         * 参数2：监听哪些消息失败
         */
        channel.addConfirmListener(ackCallback,nackCallback);
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("",queueName,null,message.getBytes(StandardCharsets.UTF_8));
        }
        // 结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布"+MESSAGE_COUNT+"条异步确认消息，耗时"+ (end - start) + "ms");
    }
}
